package com.cn.rpcframe.network;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

 
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import com.cn.rpcframe.annotation.RpcAnnotation;
import com.cn.rpcframe.registeranddiscover.ServiceRegisterAndDiscovery;
import com.cn.rpcframe.service.HelloRpc;
import com.cn.rpcframe.service.impl.HelloProcServiceImpl;
import com.cn.rpcframe.zookeeperoperation.ZookeeperServicImpl;
 

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public class RPCServer implements ApplicationContextAware, InitializingBean{
	  private static final ConcurrentHashMap<String, Class> serviceRegistry = new ConcurrentHashMap<String, Class>();
	  ServiceRegisterAndDiscovery serviceRegister;  
	  private int port;  
	    public RPCServer(int port,ServiceRegisterAndDiscovery serviceRegister){  
	        this.port = port;  
	        this.serviceRegister=serviceRegister;
	    }  
	    
	    public void start(){  
	    	 EventLoopGroup bossGroup = new NioEventLoopGroup();  
	         EventLoopGroup workerGroup = new NioEventLoopGroup();  
	    	try{
	    	
	         ServerBootstrap serverBootstrap = new ServerBootstrap().group(bossGroup, workerGroup).channel(
	        		 NioServerSocketChannel.class).localAddress(port).childHandler(new ChannelInitializer<SocketChannel>(){

						@Override
						protected void initChannel(SocketChannel ch)
								throws Exception {
							// TODO Auto-generated method stub
							ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));    
                            pipeline.addLast(new LengthFieldPrepender(4));    
                            pipeline.addLast("encoder", new ObjectEncoder());      
                            pipeline.addLast("decoder", (ChannelHandler) new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));    
                             pipeline.addLast(new InvokerHandler(serviceRegistry )); 
						}
	        			 
	        		 }).option(ChannelOption.SO_BACKLOG, 128)       
	                    .childOption(ChannelOption.SO_KEEPALIVE, true); 
	         ChannelFuture future = serverBootstrap.bind(port).sync();      
	            System.out.println("Server start listen at " + port );    
	            future.channel().closeFuture().sync();    
	    	}catch(Exception e){
	    		e.printStackTrace();
	    		  bossGroup.shutdownGracefully();    
	              workerGroup.shutdownGracefully();  
	    	}
	    }
	    
	    public void register(Class serviceInterface, Class impl) {
	        serviceRegistry.put(serviceInterface.getName(), impl);
	    }
	    public void registerZooKeeper(Class serviceInterface, String host,String port,String zookeeperHost) {
/*	    	ServiceRegisterAndDiscovery t = new ZookeeperServicImpl(zookeeperHost);
			t.register(serviceInterface.getName()+"-"+host+"-"+port);*/
	    	serviceRegister.register(serviceInterface.getName()+"-"+host+"-"+port);
	    }
	    
/*	    public static void main(String[] args) throws Exception {    
	        int port;    
	        if (args.length > 0) {    
	            port = Integer.parseInt(args[0]);    
	        } else {    
	            port = 8080;    
	        }    
	        RPCServer server = new RPCServer(port);
	        server.register(HelloRpc.class, HelloProcServiceImpl.class);
	        //将服务名字和自身的服务地址 发布到zookeeper 
	        server.registerZooKeeper(HelloRpc.class, "127.0.0.1","8080", "127.0.0.1:2181");
	        server.start();    
	    }*/
	    
	    public void startServer(){
	      
	         register(HelloRpc.class, HelloProcServiceImpl.class);
	        //将服务名字和自身的服务地址 发布到zookeeper 
	         registerZooKeeper(HelloRpc.class, "127.0.0.1","8080", "127.0.0.1:2181");
	         start();
	    }

		@Override
		public void afterPropertiesSet() throws Exception {
			// TODO Auto-generated method stub
			startServer();
		}

		//扫描RpcAnnotation 注解
		@Override
		public void setApplicationContext(ApplicationContext ctx)
				throws BeansException {
			// TODO Auto-generated method stub
	        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcAnnotation.class);
	        if ( serviceBeanMap!= null && serviceBeanMap.size() > 0) {
	            for (Object serviceBean : serviceBeanMap.values()) {
	                String interfaceName = serviceBean.getClass().getAnnotation(RpcAnnotation.class).value().getName();
	               
	                serviceRegistry.put(interfaceName, serviceBean.getClass());
	                registerZooKeeper( serviceBean.getClass(),"127.0.0.1","8080", "127.0.0.1:2181");
	            }
	        }
		}    
}
